/**
 * \file Message.cc
 * \brief Message pipe IPC mechanism
 * \author Corey Tabaka
 */

/*
   Copyright 2005 Corey Tabaka

   Licensed under the Apache License, Version 2.0 (the "License");
   you may not use this file except in compliance with the License.
   You may obtain a copy of the License at

       http://www.apache.org/licenses/LICENSE-2.0

   Unless required by applicable law or agreed to in writing, software
   distributed under the License is distributed on an "AS IS" BASIS,
   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   See the License for the specific language governing permissions and
   limitations under the License.
*/

#include <platform/Memory.h>
#include <Message.h>
#include <Process.h>
#include <Sync.h>
#include <Log.h>
#include <atomic.h>
#include <panic.h>
#include <i386/mem.h>
	
DEF_NULL_REF(MessagePipe);

MessagePipe::MessagePipe(void) : senders(),
	receivers(), sendBuffer(MESSAGE_BUFFER_INITIAL_SIZE),
	replyBuffer(MESSAGE_BUFFER_INITIAL_SIZE), valid(1) {
}

MessagePipe::~MessagePipe(void) {
	Log::printf(LOG_LOW, "Finalizing MessagePipe\n");
	
	// TODO: wake and error all waiting tasks
}

bool MessagePipe::isValid(void) {
	return valid==1;
}

void MessagePipe::invalidate(void) {
#if SMP
	atomic_set(0, &valid);
#else
	valid = 0;
#endif
}
	

int MessagePipe::receive(void *data, uint32 len) {
	receivers.lock(); // queue receivers
	
	sendBuffer.read((uint8 *) data, len);
	
	// don't signal any other receivers until reply completes	
	
	return 0;
}

int MessagePipe::reply(const void *data, uint32 len) {
	replyBuffer.write((const uint8 *) data, len);
	
	receivers.unlock(); // wake the next waiting receiver
	
	return 0;
}

int MessagePipe::send(const void *sdata, uint32 slen, void *rdata, uint32 rlen) {
	senders.lock(); // queue senders
	
	sendBuffer.write((uint8 *) sdata, slen);
	replyBuffer.read((uint8 *) rdata, rlen);
	
	senders.unlock(); // signal the next waiting sender
	
	return 0;
}

int MessagePipe::receive(void *header, uint32 hlen, void *data, uint32 dlen) {
	receivers.lock(); // queue receivers
	
	sendBuffer.read((uint8 *) header, hlen, (uint8 *) data, dlen);
	
	// don't signal any other receivers until reply completes	
	
	return 0;
}
int MessagePipe::reply(const void *header, uint32 hlen, const void *data,
	uint32 dlen) {
	replyBuffer.write((const uint8 *) header, hlen, (const uint8 *) data, dlen);
	
	receivers.unlock(); // wake the next waiting receiver
	
	return 0;
}

int MessagePipe::send(const void *sheader, uint32 shlen, const void *sdata,
	uint32 slen, void *rheader, uint32 rhlen, void *rdata, uint32 rlen) {
	senders.lock(); // queue senders
	
	sendBuffer.write((const uint8 *) sheader, shlen, (const uint8 *) sdata,
		slen);
	replyBuffer.read((uint8 *) rheader, rhlen, (uint8 *) rdata, rlen);
	
	senders.unlock(); // signal the next waiting sender
	
	return 0;
}

MessageBuffer::MessageBuffer(uint32 len) : mutex(), full(0), empty(1) {
	capacity = PAGE_ROUND(len);
	data = (uint8 *) Memory::alloc(capacity);
}

MessageBuffer::~MessageBuffer(void) {
	delete data;
}

void MessageBuffer::write(const uint8 *p, uint32 len) {
	empty.wait();
	mutex.lock();
	
	checkBuffer(len);
	
	length = len;
	memcpy(data, p, len);
	
	mutex.unlock();
	full.signal();
}

void MessageBuffer::read(uint8 *p, uint32 len) {
	full.wait();
	mutex.lock();
	
	memcpy(p, data, len < length ? len : length);
	
	mutex.unlock();
	empty.signal();
}

void MessageBuffer::write(const uint8 *h, uint32 hlen, const uint8 *p,
	uint32 len) {
	empty.wait();
	mutex.lock();
	
	checkBuffer(hlen + len);
	
	length = hlen + len;
	memcpy(data, h, hlen);
	memcpy(data + hlen, p, len);
	
	mutex.unlock();
	full.signal();
}

void MessageBuffer::read(uint8 *h, uint32 hlen, uint8 *p, uint32 len) {
	full.wait();
	mutex.lock();
	
	memcpy(h, data, hlen < length ? hlen : length);
	
	if ((length - hlen) > 0) {
		memcpy(p, data + hlen, len < (length - hlen) ? len : (length - hlen));
	}
	
	mutex.unlock();
	empty.signal();
}

void MessageBuffer::checkBuffer(uint32 len) {
	// see if the buffer needs to grow
	if (capacity < len) {
		Memory::free(data, capacity);
		
		capacity = PAGE_ROUND(len);
		data = (uint8 *) Memory::alloc(capacity);
	}
}
